"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 故障&可靠性测试
Case Name   : 备节点同步后恢复主节点
Create At   : 2021/03/09
@zou_jialiang0505328126
Description :
    文件太大了，用例看云龙
Expect      :
History     :
    Modified by @wan005 2021/5/17: 调整wal_sender_timeout,规避因环境性能差导致的失败
    Modified by @wan005 2021/8/4 : 先drop再设置备机延迟 校验结果
    Modified by @wan005 2021/11/2 : 注释gs_basebackup相关
"""

import unittest
import time
from yat.test import Node
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.CommonSH import RestartDbCluster
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class RecoveryDelay(unittest.TestCase):
    commonshpri = CommonSH('PrimaryDbUser')
    nodes_tuple = ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')

    @RestartDbCluster(*nodes_tuple)
    def setUp(self):
        self.log = Logger()
        self.log.info('----Opengauss_Reliability_Dbsys_Case263.py 开始执行------')
        self.db_primary_root_node = Node(node='PrimaryRoot')
        self.db_standby_node =  \
            [Node(node='Standby1Root'), Node(node='Standby2Root')]
        self.db_standby_user_node =  \
            [Node(node='Standby1DbUser'), Node(node='Standby2DbUser')]
        self.db_primary_user_node = Node(node='PrimaryDbUser')
        self.comshsta = \
            [CommonSH('Standby1DbUser'), CommonSH('Standby2DbUser')]
        self.constant = Constant()
        self.com_pri = Common('PrimaryRoot')
        self.com_sta = [Common('Standby1Root'), Common('Standby2Root')]
        self.tb_name = 'tb_259'
        self.trigger_name = 'insert_trigger'
        self.func_name = 'tri_insert_func'

        self.log.info('====install ntp===')
        install_cmd = "yum install -y ntp && " \
                      "systemctl enable ntpd && systemctl start ntpd"
        for i in range(2):
            result = self.db_standby_node[i].sh(install_cmd).result()
            self.log.info(result)

        self.log.info('----sync time----')
        for i in range(2):
            current = self.db_primary_root_node.sh(
                "date \"+%m/%d/%Y %H:%M:%S\"").result()
            self.log.info(current)
            datecmd = f'date -s "{current}";hwclock --systohc;' \
                f'service ntpd stop;ntpdate ntp.api.bz;service ntpd start'
            self.log.info(datecmd)
            result = self.db_standby_node[i].sh(datecmd).result()
            self.log.info(result)

        result = self.commonshpri.executDbSql('show wal_sender_timeout;')
        self.log.info(f"wal_sender_timeout is {result}")
        self.wal_sender_timeout = result.strip().splitlines()[-2]

    def test_recoverydelay(self):
        sql = f"DROP TABLE IF EXISTS {self.tb_name};" \
              f"drop sequence if exists serial;"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result)
        self.assertIn("DROP SEQUENCE", result)

        self.log.info('----modify wal_sender_timeout------')
        result = self.commonshpri.executeGsguc('reload',
                                          self.constant.GSGUC_SUCCESS_MSG,
                                          'wal_sender_timeout=12s')
        self.assertTrue(result)

        self.log.info('----recovery_min_apply_delay=3min----')
        result = self.commonshpri.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            "recovery_min_apply_delay='3min'")
        self.assertTrue(result)

        self.log.info('---switch to cascade-----')
        result = self.comshsta[0].buildStandby('-M cascade_standby')
        self.assertIn(self.constant.REBUILD_SUCCESS_MSG, result)
        result = self.comshsta[0].exec_refresh_conf()
        self.assertTrue(result)

        self.log.info('------consistent---------')
        result = self.comshsta[1].check_data_consistency()
        self.assertTrue(result)
        for i in range(90):
            result = self.commonshpri.check_cascade_standby_consistency()
            if result:
                break
            time.sleep(20)
        self.assertTrue(result)

        self.log.info('-------create tb----')
        sql = f"CREATE TABLE {self.tb_name}(num int)" \
            f"PARTITION BY RANGE(num)" \
            f"(PARTITION num1 VALUES LESS THAN(10000), " \
            f"PARTITION num2 VALUES LESS THAN(20000)," \
            f"PARTITION num3 VALUES LESS THAN(300000));"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, result)

        self.log.info('------create func-----')
        sql = f"CREATE OR REPLACE FUNCTION {self.func_name}() " \
            f"RETURNS TRIGGER AS  " \
            f"\$\$   " \
            f"DECLARE   " \
            f"BEGIN  " \
            f"INSERT INTO {self.tb_name} " \
            f"VALUES(NEW.id1);  " \
            f"RETURN NEW;      " \
            f"END           " \
            f"\$\$ LANGUAGE PLPGSQL;"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        self.assertIn('CREATE FUNCTION', result)

        sql = f" CREATE TRIGGER {self.trigger_name} " \
            f"  BEFORE INSERT ON {self.tb_name}" \
            f"  FOR EACH ROW " \
            f" EXECUTE PROCEDURE {self.func_name}();"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        self.assertIn('CREATE TRIGGER', result)

        sql = f"CREATE SEQUENCE serial START 101 CACHE 20;"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        self.assertIn('CREATE SEQUENCE', result)

        self.log.info('---check nodes----')
        sql = f"select sysdate;select tgname from pg_trigger;"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        self.assertIn(self.trigger_name, result)
        for i in range(2):
            result = self.comshsta[i].executDbSql(sql)
            self.log.info(result)
            self.assertNotIn(self.trigger_name, result)
        sql = f"select sysdate;\sf {self.func_name}"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        self.assertIn(self.func_name, result)
        for i in range(2):
            result = self.comshsta[i].executDbSql(sql)
            self.log.info(result)
            self.assertIn(self.constant.NOT_EXIST, result)
        sql = f"select sysdate;\ds"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        self.assertIn('serial', result)
        for i in range(2):
            result = self.comshsta[i].executDbSql(sql)
            self.log.info(result)
            self.assertNotIn('serial', result)
        time.sleep(240)
        sql = f"select sysdate;select tgname from pg_trigger;"
        for i in range(2):
            result = self.comshsta[i].executDbSql(sql)
            self.log.info(result)
            self.assertIn(self.trigger_name, result)
        sql = f"select sysdate;\sf {self.func_name}"
        for i in range(2):
            result = self.comshsta[i].executDbSql(sql)
            self.log.info(result)
            self.assertIn(self.func_name, result)
        sql = f"select sysdate;\ds"
        for i in range(2):
            result = self.comshsta[i].executDbSql(sql)
            self.log.info(result)
            self.assertIn('serial', result)

    def tearDown(self):
        self.log.info('------teardown-----')

        self.log.info('-----recovery_min_apply_delay---')
        result = self.commonshpri.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            'recovery_min_apply_delay=0')
        self.log.info(result)

        self.log.info('--------recovery-------')
        self.comshsta[0].buildStandby('-M standby')
        result = self.comshsta[0].exec_refresh_conf()
        self.log.info(result)

        self.log.info('-----------restart db-----------')
        result = self.commonshpri.stopDbCluster()
        self.log.info(result)
        result = self.commonshpri.startDbCluster()
        self.log.info(result)

        self.log.info('--------------delete------------------')
        sql = f"DROP FUNCTION IF EXISTS {self.func_name}() CASCADE;"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        sql = f"DROP table IF EXISTS {self.tb_name};"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)
        sql = f"DROP sequence IF EXISTS serial;"
        result = self.commonshpri.executDbSql(sql)
        self.log.info(result)

        self.log.info('----check wether build or not----')
        result = self.commonshpri.check_whether_need_build()
        if result:
            for i in range(2):
                self.comshsta[i].executeGsctl(
                    'build', self.constant.BUILD_SUCCESS_MSG,
                    '-b full')

        self.log.info('还原wal_sender_timeout')
        result = self.commonshpri.executeGsguc('reload',
                                          self.constant.GSGUC_SUCCESS_MSG,
                                          f'wal_sender_timeout='
                                          f'{self.wal_sender_timeout}')
        self.log.info(result)
        self.log.info('Opengauss_Reliability_Dbsys_Case263.py end')